package com.bw.gmall.realtime.utils;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Flinkcdc1 {

    public static SingleOutputStreamOperator<String> mysqlCDC(StreamExecutionEnvironment env, String database, String... tables) throws Exception {
        //1.获取执行环境



        String tableList = "";
        for (String table : tables) {
            tableList += database + "." + table + ",";
        }

        //2.通过FlinkCDC 构建SourceFunction并读取数据
        MySqlSource<String> sourceFunction = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(10000)
                .username("root")
                .password("root")
                .databaseList(database)//可以同时读多个库
                .tableList(tableList.substring(0, tableList.length() - 1))
                //可选配置项,如果不指定该参数,则会读取上一个配置下的所有表的数据，注意：指定的时候需要使用"db.table"的方式
                .deserializer(new CustomerDeserialization())
                //initial 初始化，
                //.startupOptions(StartupOptions.initial())
                .startupOptions(StartupOptions.initial())  //latest 是直接读最新的数据，而不会去打印老的数据
                .build();
//        SingleOutputStreamOperator<String> streamSource = env.addSource(sourceFunction);
        DataStreamSource<String> stream = env.fromSource(
                sourceFunction, WatermarkStrategy.noWatermarks(), "MySQL Source"
        );
                        /*.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                return JSONObject.parseObject(value).getJSONObject("after").toJSONString();
            }
        })*/;
        //3.打印数据并将数据写入Kafka
//        stream.print();
        //4.启动任务
        return stream;
    }
}
